package org.ala.tree.lsm.sstable;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.Iterator;
import java.util.List;
import java.util.PriorityQueue;
import java.util.UUID;

import org.ala.tree.lsm.KV;
import org.ala.tree.lsm.LSMConfig;
import org.ala.tree.lsm.sstable.block.DataBlock;
import org.ala.tree.lsm.sstable.block.MetaBlock;
import org.ala.tree.lsm.sstable.block.TailBlock;
import org.ala.tree.lsm.tools.FileTools;

/**
 * SSTable数据合并
 *
 * @author ala
 * @date 2025年5月29日
 */
public class SSTableMerger<V> {

	
	/**
	 * 	检测当前是否要合并
	 * 	<p>	目录下的文件，要么在值域以外[2, countMerge]，要么等于countMerge
	 */
	protected boolean checkMerge(LSMConfig config, int level) {
		int cnt = FileTools.fileCount(String.format("%s/%s%d", config.ROOT_DIR, config.levelPrefix, level));
		return cnt >= Math.max(2, config.countLevel);
	}
	/**
	 * 	尝试合并
	 * 	<p>	这里其实有问题，锁没有考虑。同一层同一时刻多次触发合并时，SSTable数量会出现幻读
	 * 	@throws IOException 
	 */
	@SuppressWarnings("unchecked")
	public void tryMerge(LSMConfig config, int level, Class<V> vclass, SSTableWriter<V> writer) throws IOException {
		if (!checkMerge(config, level)) {return;}
		
		List<Path> files = FileTools.filesAll(String.format("%s/%s%d", config.ROOT_DIR, config.levelPrefix, level), "sstable");
		SSTableIterator<V>[] iterators = new SSTableIterator[files.size()];
		for (int i = 0 ; i < files.size() ; i++) {
			iterators[i] = new SSTableIterator<>(files.get(i), vclass);
		}
		_merge(config, level, iterators, writer);
		//	成功合并本层后，清除刚才合并的文件
		files.forEach(p -> {
			String file = p.toFile().getAbsolutePath();
			boolean r = FileTools.fileDel(file);
			System.out.println("合并后清除文件 r:" + r + " file: " + file);
		});
		
		//	尝试上一级是否能合并
		tryMerge(config, level+1, vclass, writer);
	}
	protected void _merge(LSMConfig config, int level, SSTableIterator<V>[] iterators, SSTableWriter<V> writer) throws IOException {
		String id = UUID.randomUUID().toString();
		String dir = String.format("%s/%s%d", config.ROOT_DIR, config.levelPrefix, level+1);
		FileTools.tryMakeDir(dir);
		String file = String.format("%s/%s.sstable", dir, id);
		
		PriorityQueue<Integer> heap = new PriorityQueue<>(iterators.length, (i1, i2) -> {
			KV<V> kv1 = iterators[i1].current(), kv2 = iterators[i2].current();
			return kv1.getKey().compareTo(kv2.getKey());
		});
		int size = 0;
		for (int i = 0 ; i < iterators.length ; i++) {
			if (iterators[i].hasNext()) {
				heap.offer(i);
				size += iterators[i].dataCount;
			}
		}
		
		Iterator<KV<V>> iterator = new Iterator<>() {
			public boolean hasNext() {
				return !heap.isEmpty();
			}
			public KV<V> next() {
				int i = heap.poll();
				KV<V> kv = iterators[i].next();
				kv.timestamp(iterators[i].getTimestamp());
				if (iterators[i].hasNext()) {
					heap.offer(i);
				}
				return kv;
			}
		};
		
		writer.save(file, size, config.blockCapacity, iterator);
	}
	
	
	
	/**
	 * 	SSTable文件迭代器
	 */
	public static class SSTableIterator<V> implements Iterator<KV<V>> {
		protected FileChannel fileChannel;
		protected TailBlock tb;
		/**
		 * 	键值对总数，数据段总长，DataBlock容量，已读数据长度
		 */
		protected int dataCount = -1, 
					  dataLen = -1,
					  dataBlockCapacity = -1,
					  read = 0;
		/**
		 * 	写文件时间戳
		 */
		protected long timestamp;
		/**
		 * 	当前数据块
		 */
		protected DataBlock<V> dataBlock;
		/**
		 * 	数据块值类型
		 */
		protected Class<V> vclass;
		public SSTableIterator(Path path, Class<V> vclass) throws IOException {
			this.vclass = vclass;
			
			fileChannel = FileChannel.open(path, StandardOpenOption.READ);
			//	读文件尾
			MappedByteBuffer tailBuf = fileChannel.map(MapMode.READ_ONLY, fileChannel.size() - TailBlock.LEN, TailBlock.LEN);
			tb = TailBlock.decode(tailBuf);
			//	读元区
			MappedByteBuffer metaBuf = fileChannel.map(MapMode.READ_ONLY, tb.getMetaOffset(), tb.getMetaLen());
			MetaBlock mb = MetaBlock.decode(metaBuf, false);
			
			//	数据段总长，每个DataBlock容量
			dataCount = mb.getCount();
			dataLen = tb.getIndexOffset();
			dataBlockCapacity = mb.getDataBlockCapacity();
			timestamp = mb.getTimestamp();
		}
		public boolean hasNext() {
			if (dataBlock == null) {
				try {
					dataBlock = readDataBlock(vclass);
					if (dataBlock == null) {return false;}
				} catch (IOException e) {
					e.printStackTrace();
					return false;
				}
			}
			return true;
		}
		public KV<V> current() {
			return dataBlock.current();
		}
		public KV<V> next() {
			KV<V> kv = dataBlock.next();
			if (!dataBlock.hasNext()) {
				dataBlock = null;
			}
			return kv;
		}
		/**
		 * 	读一个DataBlock
		 * @throws IOException 
		 */
		protected DataBlock<V> readDataBlock(Class<V> vclass) throws IOException {
			if (read + 4 >= dataLen) {return null;}
			
			//	读DataBlock长度
			read += 4;
			ByteBuffer cntBuf = ByteBuffer.allocate(4);
			fileChannel.read(cntBuf);
			cntBuf.flip();
			int len = cntBuf.getInt();

			if (read + len >= dataLen) {return null;}
			//	根据长度读DataBlock
			ByteBuffer dataBuf = ByteBuffer.allocate(len);
			fileChannel.read(dataBuf);
			dataBuf.flip();
			
			//	累加已读长度
			read += len;
			DataBlock<V> db = DataBlock.decode(dataBlockCapacity, dataBuf, vclass);
			db.resetIteratprPos();
			return db;
		}
		public long getTimestamp() {
			return timestamp;
		}
		public void setTimestamp(long timestamp) {
			this.timestamp = timestamp;
		}
	}
	
}
